package cn.iocoder.boot.MLlib;

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.feature.VectorAssembler;
import org.apache.spark.ml.regression.LinearRegression;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 这是一个用于机器学习模型训练和保存的服务类
 * 基于 Apache Spark MLlib 实现线性回归模型的训练和持久化
 */
@Service
//服务类，用于训练机器学习模型并将其保存。
public class ModelTrainingService {

    // 使用 Spring 自动注入 SparkSession，用于与 Spark 进行交互
    @Autowired
    private SparkSession sparkSession;

    /**
     * 训练并保存线性回归模型
     * @param trainingDataPath 训练数据的 Parquet 文件路径
     * @param modelSavePath 模型保存路径
     * @throws IOException 如果保存模型时发生 I/O 错误
     */
    public void trainAndSaveModel(String trainingDataPath, String modelSavePath) throws IOException {
        // 加载训练数据
        // SparkSession.read().parquet() 从 Parquet 文件中读取数据，生成 Dataset<Row> 对象
        Dataset<Row> trainingData = sparkSession.read().parquet(trainingDataPath);

        // 定义特征列和标签列 这里假设数据中有两列特征列 "feature1" 和 "feature2"
        List<String> featureColumns = new ArrayList<>();
        featureColumns.add("feature1");
        featureColumns.add("feature2");

        // 创建 VectorAssembler，将多列特征组合成一个特征向量
        // setOutputCol("features") 指定输出的特征向量列名为 "features"
        VectorAssembler assembler = new VectorAssembler()
                .setInputCols(featureColumns.toArray(new String[0]))  // 设置输入的特征列
                .setOutputCol("features");                          // 设置输出的特征向量列名

        // 创建线性回归模型,setLabelCol("label") 指定标签列名称
        // setFeaturesCol("features") 指定特征列名称（默认即为 "features"）
        LinearRegression lr = new LinearRegression()
                .setLabelCol("label")    // 设置标签列名
                .setFeaturesCol("features");  // 设置特征列名

        // 创建 Pipeline（管道）
        // Pipeline 是一个工作流，包含多个按顺序执行的 PipelineStage（阶段）
        // 这里包含两个阶段：特征组装和线性回归训练
        Pipeline pipeline = new Pipeline()
                .setStages(new PipelineStage[]{assembler, lr});  // 设置管道的阶段

        // 使用训练数据拟合管道，训练模型
        // Pipeline.fit() 方法会按顺序执行所有阶段，最终生成 PipelineModel
        PipelineModel model = pipeline.fit(trainingData);

        // 保存训练好的模型,PipelineModel.save() 方法将模型序列化并保存到指定路径
        // 保存的模型可以被重新加载用于预测
        model.save(modelSavePath);
    }
}






//在Spring Boot项目中，ModelTrainingService类负责机器学习模型的训练和保存，以下是逐行讲解：
//
//        1. 导入所需的Apache Spark机器学习库和Spring框架相关类：
//
//        ```java
//import org.apache.spark.ml.Pipeline;
//import org.apache.spark.ml.evaluation.RegressionEvaluator;
//import org.apache.spark.ml.feature.VectorAssembler;
//import org.apache.spark.ml.regression.LinearRegression;
//import org.apache.spark.sql.Dataset;
//import org.apache.spark.sql.Row;
//import org.apache.spark.sql.SparkSession;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Service;
//```
//
//        2. 使用`@Service`注解标记此类为Spring服务组件，使其被Spring容器管理：
//
//        ```java
//@Service
//public class ModelTrainingService {
//```
//
//        3. 通过`@Autowired`注解自动装配`SparkSession`，用于与Spark交互：
//
//            ```java
//    @Autowired
//    private SparkSession sparkSession;
//```
//
//        4. 定义`trainAndSaveModel`方法，接收训练数据路径和模型保存路径参数：
//
//            ```java
//    public void trainAndSaveModel(String trainingDataPath, String modelSavePath) {
//```
//
//        5. 使用`sparkSession.read().parquet(trainingDataPath)`加载Parquet格式的训练数据：
//
//```java
//        Dataset<Row> trainingData = sparkSession.read().parquet(trainingDataPath);
//```
//
//        6. 创建特征列列表，指定参与模型训练的特征列名称：
//
//```java
//        List<String> featureColumns = new ArrayList<>();
//        featureColumns.add("feature1");
//        featureColumns.add("feature2");
//```
//
//        7. 初始化`VectorAssembler`，将多列特征合并为一个特征向量：
//
//```java
//        VectorAssembler assembler = new VectorAssembler()
//                .setInputCols(featureColumns.toArray(new String[0]))
//                .setOutputCol("features");
//```
//
//        8. 创建线性回归模型实例，指定标签列和特征列：
//
//```java
//        LinearRegression lr = new LinearRegression()
//                .setLabelCol("label")
//                .setFeaturesCol("features");
//```
//
//        9. 构建Pipeline，包含特征组装和模型训练两个阶段：
//
//```java
//        Pipeline pipeline = new Pipeline()
//                .setStages(new PipelineStage[]{assembler, lr});
//```
//
//        10. 使用训练数据拟合Pipeline，生成训练好的模型：
//
//```java
//        PipelineModel model = pipeline.fit(trainingData);
//```
//
//        11. 调用`model.save(modelSavePath)`将训练好的模型保存到指定路径：
//
//```java
//        model.save(modelSavePath);
//```
//
//
//
//**注意**：建议在方法中添加异常处理和资源释放逻辑，以确保程序的健壮性和资源的正确管理。